package org.sg20.labelpropagation.lpainiter;




import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.sg20.utils.LpaDotVal;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class LpaIniterMapper extends Mapper<Text, Text, Text, Text> {

    private Map<String, LpaDotVal> initDots;
    private int labelNum;
    private Text outKey;
    private Text outVal;

    @Override
    protected void setup(Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        super.setup(context);
        this.labelNum = context.getConfiguration().getInt("label-num", 1);
        this.initDots = new HashMap<>();
        this.outKey = new Text();
        this.outVal = new Text();
        URI[] cacheFiles = context.getCacheFiles();
        if (cacheFiles != null && cacheFiles.length > 0) {
            /*
                读入初始的,已经带有标签的点集
                格式为"人名\t 标签"
             */
            {
                Path cache = new Path(cacheFiles[0]);
                String line;
                String[] tokens;
                FileSystem fs = FileSystem.get(context.getConfiguration());
                InputStreamReader is = new InputStreamReader(fs.open(cache));
                BufferedReader labelReader = new BufferedReader(is);
                try {
                    {
                        while ((line = labelReader.readLine()) != null) {
                            tokens = line.split("\t", 2);
                            String name = tokens[0];
                            int label = Integer.parseInt(tokens[1]);
                            initDots.put(name, new LpaDotVal(labelNum, label));
                        }
                    }
                } finally {
                    labelReader.close();
                }
            }
        }
    }


    @Override
    protected void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        /**
         * 唐僧 [悟空.0.5|八戒，0. 5]-> 唐僧 [pr0,pr1,...] [悟空.0.5|八戒，0. 5]
         * key:唐僧
         * val
         */
        String name = key.toString();
        LpaDotVal dotVal;
        String res;
        if ((dotVal = initDots.get(name)) == null) {
            res = new LpaDotVal(labelNum).toString(); //"[0,0,...]"
        } else {
            res = dotVal.toString();
        }
        res = res + ' ' + value.toString();
        outVal.set(res);
        outKey = key;
        context.write(outKey, outVal);
    }
}